package com.lagou.client;

import com.alibaba.fastjson.JSONObject;
import com.lagou.service.UserService;
import com.lagou.service.com.lagou.util.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.util.StringUtils;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RpcConsumer {

    //创建线程池对象
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

//    private static UserClientHandler userClientHandler = new UserClientHandler();

    private static String rootPath = "/zdy-rpc";

//    private TreeMap<Long,UserService> map = new TreeMap<>();

    private Map<String, UserService> sMap = new HashMap();

    private Map<String,Long> map = new LinkedHashMap<>();

    ZkClient zkClient = new ZkClient("127.0.0.1:2181");

    // treemap 中存放 key: 调用时间 value：代理类实例


    public Map<String, UserService> getsMap() {
        return sMap;
    }

    public void setsMap(Map<String, UserService> sMap) {
        this.sMap = sMap;
    }

    public Map<String, Long> getMap() {
        return map;
    }

    public void setMap(Map<String, Long> map) {
        this.map = map;
    }

    public static Bootstrap initBootstrap(UserClientHandler userClientHandler) {
        Bootstrap bootstrap = new Bootstrap();
        EventLoopGroup group = new NioEventLoopGroup();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new StringDecoder());
//                        pipeline.addLast(new RpcEncoder(RpcRequest.class,new JSONSerializer()));
//                        pipeline.addLast(new RpcDecoder(RpcRequest.class,new JSONSerializer()));
//                        pipeline.addLast(new JsonObjectDecoder());
                        pipeline.addLast(userClientHandler);
                    }
                });
        return bootstrap;
    }

    //1.创建一个代理对象 providerName：UserService#sayHello are you ok?
    public Object createProxy(final Class<?> serviceClass, RpcRequest request, UserClientHandler userClientHandler, String nodeName){
        //借助JDK动态代理生成代理对象
        return  Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                //（1）调用初始化netty客户端的方法


//                initClient();

                request.setParameters(args);
                Class<?> clazz[] = new Class[args.length];
                for (int i = 0; i < args.length; i++) {
                    clazz[i] = args[i].getClass();
                }
                request.setParameterTypes(clazz);
                // 设置参数
//                userClientHandler.setPara(providerName+args[0]);
                String s = JSONObject.toJSON(request).toString();


                userClientHandler.setPara(s);

                // 去服务端请求数据
                long l = System.currentTimeMillis();
                Object o = executor.submit(userClientHandler).get();
                // 节点数据格式 ： 上次调用耗费时间，调用时间戳
                zkClient.writeData(rootPath+"/"+nodeName, System.currentTimeMillis()-l + "," + System.currentTimeMillis());
                return o;
            }
        });


    }



    //2.初始化netty客户端
    public  void initClient() throws InterruptedException {


//        EventLoopGroup group = new NioEventLoopGroup();

        // 首先从rootpath 节点下获取子节点
        List<String> children = zkClient.getChildren(rootPath);
        for (String child : children) {
            UserClientHandler userClientHandler = new UserClientHandler();
            // 与节点下的服务建立连接
            String[] str = child.split(":");



            RpcConsumer.initBootstrap(userClientHandler).connect(str[0], Integer.parseInt(str[1])).sync();
            System.out.println("建立连接：" + str);
            // 生成代理类对象放入map中返回
            UserService proxy = (UserService) createProxy(UserService.class, ClientBootStrap.request, userClientHandler,str[0] + ":" + str[1]);

            // 注册dataChange监听
            zkClient.subscribeDataChanges(rootPath + "/" + str[0] + ":" + str[1], new IZkDataListener() {
                @Override
                public void handleDataChange(String s, Object o) throws Exception {
                    // 维护 代理类实例map
                    handleData(String.valueOf(o), rootPath + "/" + str[0] + ":" + str[1]);
                }

                @Override
                public void handleDataDeleted(String s) throws Exception {
                    handleData("", rootPath + "/" + str[0] + ":" + str[1]);
                }
            });

            // map 初始化
            sMap.put(rootPath + "/" + str[0] + ":" + str[1], proxy);
            handleData("", rootPath + "/" + str[0] + ":" + str[1]);
        }

        // 从zookeeper中监听 节点/zdy-rpc 下的节点列表并建立连接
        zkClient.subscribeChildChanges(rootPath, new IZkChildListener() {
            @Override
            public void handleChildChange(String s, List<String> list) throws Exception {
                for (String s1 : list) {
                    UserClientHandler userClientHandler = new UserClientHandler();
                    String[] str = s1.split(":");
                    initBootstrap(userClientHandler).connect(str[0], Integer.parseInt(str[1])).sync();
                    System.out.println("建立连接：" + str[0]+":" + str[1]);
                    UserService proxy = (UserService) createProxy(UserService.class, ClientBootStrap.request, userClientHandler,str[0]+":" + str[1]);

                    // 注册dataChange监听
                    zkClient.subscribeDataChanges(rootPath + "/" + str[0] + ":" + str[1], new IZkDataListener() {
                        @Override
                        public void handleDataChange(String s, Object o) throws Exception {
                            // 维护 代理类实例map
                            handleData(String.valueOf(o), rootPath + "/" + str[0] + ":" + str[1]);
                        }

                        @Override
                        public void handleDataDeleted(String s) throws Exception {
                            handleData("", rootPath + "/" + str[0] + ":" + str[1]);
                        }
                    });
                    sMap.put(rootPath + "/" + str[0] + ":" + str[1], proxy);

                }
            }


        });

        // 注册监听dataChange 事件

//        bootstrap1.connect("127.0.0.1",8010).sync();

    }

    void handleData(String data, String pathName){
        if(StringUtils.isEmpty(data)){
            map.put(pathName, 0L);
        } else {
            String[] s2 = data.split(",");
            long l = Long.parseLong(s2[1]);
            long l1 = Long.parseLong(s2[0]);
            // 如果最后一次调用时间，距离现在大于5S则调用时间清空
            if(System.currentTimeMillis()-l > 5000){
                map.put(pathName, 0L);
                map = sortMapByValue(map);
                return;
            }
            map.put(pathName,l1);
            map = sortMapByValue(map);
        }
        map = sortMapByValue(map);
    }

    public Map<String, Long> sortMapByValue(Map<String, Long> oriMap) {
        if (oriMap == null || oriMap.isEmpty()) {
            return null;
        }
        Map<String, Long> sortedMap = new LinkedHashMap<String, Long>();
        List<Map.Entry<String, Long>> entryList = new ArrayList<Map.Entry<String, Long>>(
                oriMap.entrySet());
        Collections.sort(entryList, new Comparator<Map.Entry<String, Long>>() {
            @Override
            public int compare(Map.Entry<String, Long> o1, Map.Entry<String, Long> o2) {
                return o1.getValue().compareTo(o2.getValue());
            }
        });

        Iterator<Map.Entry<String, Long>> iter = entryList.iterator();
        Map.Entry<String, Long> tmpEntry = null;
        while (iter.hasNext()) {
            tmpEntry = iter.next();
            sortedMap.put(tmpEntry.getKey(), tmpEntry.getValue());
        }
        return sortedMap;

    }

}
